package com.fashopcosmos.preferencias.mapreduce;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import com.fashopcosmos.preferencias.beans.GrupoSexoEdad;
import com.fashopcosmos.preferencias.beans.ListClienteProducto;
import com.fashopcosmos.preferencias.beans.PreferenceCustomerBean;

public class PreferenceReduce extends MapReduceBase
	implements Reducer<Text, PreferenceCustomerBean, Text, Text>{
	
//	private final int EDAD_10=10;
//	private final int EDAD_20=20;
//	private final int EDAD_30=30;
//	private final int EDAD_40=40;
//	private int contadorM	=	0;
//	private int contadorF	=	0;
	@Override
	public void reduce(Text key, Iterator<PreferenceCustomerBean> value,
			OutputCollector<Text, Text> output, Reporter reporter)
	throws IOException {
		HashMap<String, ListClienteProducto> hmp = new HashMap<String, ListClienteProducto>();
		while(value.hasNext()){
			PreferenceCustomerBean c=value.next();
			String producto=c.getProducto();
			if(hmp.containsKey(producto)){
				ListClienteProducto ldata= hmp.get(producto);
				GrupoSexoEdad gse=new GrupoSexoEdad();
				gse.setEdad(c.getEdad());
				gse.setSexo(c.getSexo());
				gse.setContar(1);
				GrupoSexoEdad tmpGse=ldata.buscar(gse);
				if(tmpGse!=null){
					gse.setContar(tmpGse.getContar()+1);
				}
				ldata.setAddDatos(gse);
				hmp.put(producto, ldata);
			}
			else{
				ListClienteProducto ldata=new ListClienteProducto();
				GrupoSexoEdad gse=new GrupoSexoEdad();
				gse.setEdad(c.getEdad());
				gse.setSexo(c.getSexo());
				gse.setContar(1);
				ldata.setAddDatos(gse);
				hmp.put(producto, ldata);
			}			
		} 
		
		Iterator it = hmp.entrySet().iterator();
		while(it.hasNext()){
			Entry<String,ListClienteProducto> e=(Entry<String,ListClienteProducto>)it.next();
			String sKey=e.getKey();
			String resultado;
			ListClienteProducto listGSE=e.getValue();
			List<GrupoSexoEdad> lGse = listGSE.getDataCliente();
			for(int i=0; i<lGse.size();i++){
				GrupoSexoEdad tmpgse=lGse.get(i);
				resultado= sKey +" - "+tmpgse.getSexo() +" - " +tmpgse.getContar();
				output.collect(key, new Text(resultado));
			}			
		}
		
	}
}
